﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Confluent.Kafka;
using Newtonsoft.Json;

namespace ConsoleAppKafkaTest
{
    class KafkaConsumerDemo
    {
        public static void TestConsumer()
        {
            ConsumerConfig conConfig = new ConsumerConfig();
            conConfig.BootstrapServers = "127.0.0.1:9092";
            conConfig.GroupId = Guid.NewGuid().ToString();

            conConfig.AutoOffsetReset = AutoOffsetReset.Latest; //AutoOffsetReset.Earliest;
            conConfig.EnableAutoCommit = false;

            ConsumerBuilder<string, object> conBuilder = new ConsumerBuilder<string, object>(conConfig);

            conBuilder.SetErrorHandler((Null, r) =>
            {
                if (r.IsError)
                {
                    Console.WriteLine($"接收数据发生错误。。。");
                }
            });
            conBuilder.SetValueDeserializer(new KafkaDeserializerConverter());
            IConsumer<string, object> consumer = conBuilder.Build();
            consumer.Subscribe("haha1");

            while (true)
            {
                ConsumeResult<string, object> result = consumer.Consume();
                Console.WriteLine($"recive message:[{result.Message.Key}:{result.Message.Value}]");
                consumer.Commit(result);//手动提交数据
                Console.WriteLine("---------------------------------------");
            }
        }

    }

    public class KafkaDeserializerConverter : IDeserializer<object>
    {
        public object Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
        {
            if (isNull)
            {
                return null;
            }
            string jsonStr = Encoding.UTF8.GetString(data.ToArray());
            try
            {
                return JsonConvert.DeserializeObject(jsonStr);
            }
            catch (Exception)
            {
                return jsonStr;
            }

        }
    }
}
